package hzmg.akkacloud.server.remote.transport.starter;

import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.cluster.Cluster;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hzmg.akkacloud.common.AkkaConstant;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.utils.NetUtils;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import hzmg.akkacloud.server.common.AkkaCloudServerConfigKey;
import hzmg.akkacloud.server.common.utils.PropertyUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.*;

/**
 * 服务端 ActorSystem 启动器
 *
 * @author zbw
 * @since 2022/7/17
 */
@Slf4j
public class AkkaStarter {

    private static final String AKKA_PATH = "akka://%s@%s/user/%s";
    public static ActorSystem actorSystem;
    @Getter
    private static String actorSystemAddress;

    public static void init() {

        Stopwatch stopwatch = Stopwatch.createStarted();
        log.info("[AkkaCloud] AkkaCloud's akka system start to bootstrap...");

        // 解析配置文件
        Properties properties = PropertyUtils.getProperties();

        int port = Integer.parseInt(properties.getProperty(AkkaCloudServerConfigKey.AKKA_PORT, String.valueOf(AkkaConstant.SERVER_DEFAULT_AKKA_PORT)));
        String portFromJVM = System.getProperty(AkkaCloudServerConfigKey.AKKA_PORT);
        if (StringUtils.isNotEmpty(portFromJVM)) {
            log.info("[AkkaCloud] use port from jvm params: {}", portFromJVM);
            port = Integer.parseInt(portFromJVM);
        }

        // 启动 ActorSystem
        Map<String, Object> overrideConfig = Maps.newHashMap();
        String localIP = NetUtils.getLocalHost();
        overrideConfig.put("akka.remote.artery.canonical.hostname", localIP);
        overrideConfig.put("akka.remote.artery.canonical.port", port);
        // 设置角色
        Set<String> useRoles = new HashSet<>(Collections.singletonList(RemoteConstant.SERVER_CONSOLE_ROLE));
        overrideConfig.put("akka.cluster.roles", useRoles);
        actorSystemAddress = localIP + ":" + port;
        log.info("[AkkaCloud] akka-remote server address: {}", actorSystemAddress);

        Config akkaBasicConfig = ConfigFactory.load(RemoteConstant.SERVER_AKKA_CONFIG_NAME);
        Config akkaFinalConfig = ConfigFactory.parseMap(overrideConfig).withFallback(akkaBasicConfig);
        actorSystem = ActorSystem.create(RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, akkaFinalConfig);
        //创建集群
        Cluster cluster = Cluster.get(actorSystem);
        //加入种子节点
        List<String> everyRegistry=Arrays.asList(properties.getProperty(AkkaCloudServerConfigKey.EVERY_REGISTRY).split(","));
        List<Address> nodeList = Lists.newLinkedList();
        everyRegistry.forEach(registryAddress -> {
            String[] ipWithPort = registryAddress.split(":");
            nodeList.add(new Address("akka", RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, ipWithPort[0], Integer.parseInt(ipWithPort[1])));
        });
        cluster.joinSeedNodes(nodeList);
        log.info("[Akka cluster] cluster address:{}",cluster.selfAddress());
        log.info("[AkkaCloud] AkkaCloud's akka system started successfully, using time {}.", stopwatch);
    }

    /**
     * 获取 ServerActor 的 ActorSelection
     *
     * @param address IP:port
     * @return ActorSelection
     */
    public static ActorSelection getFriendActor(String address) {
        String path = String.format(AKKA_PATH, RemoteConstant.SERVER_ACTOR_SYSTEM_NAME, address, RemoteConstant.SERVER_FRIEND_ACTOR_NAME);
        return actorSystem.actorSelection(path);
    }

    public static ActorSelection getWorkerActor(String address) {
        String path = String.format(AKKA_PATH, RemoteConstant.WORKER_ACTOR_SYSTEM_NAME, address, RemoteConstant.WORKER_ACTOR_NAME);
        return actorSystem.actorSelection(path);
    }
}
